/*
 * Copyright (c) Jipzingking 2016.
 */

package sdfs.namenode;

import sdfs.client.DataNodeStub;
import sdfs.datanode.DataNode;
import sdfs.filetree.BlockInfo;
import sdfs.filetree.FileNode;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.ReadOnlyBufferException;
import java.nio.channels.SeekableByteChannel;
import java.util.*;

public class SDFSFileChannel implements SeekableByteChannel, Flushable, Serializable {
    private static final long serialVersionUID = 6892411224902751501L;
    private final UUID uuid; //File uuid
    private int fileSize; //Size of this file
    private int blockAmount; //Total block amount of this file
    private final FileNode fileNode;
    private final boolean isReadOnly;
    private final Map<Integer, byte[]> dataBlocksCache = new HashMap<>(); //BlockNumber to DataBlock cache. byte[] or ByteBuffer are both acceptable.
    private NameNode nameNode;
    private boolean isOpen=true;

    SDFSFileChannel(NameNode nameNode ,UUID uuid, int fileSize, int blockAmount, FileNode fileNode, boolean isReadOnly) {
        this.uuid = uuid;
        this.fileSize = fileSize;
        this.blockAmount = blockAmount;
        this.fileNode = fileNode;
        this.isReadOnly = isReadOnly;
        this.nameNode =nameNode;
        init();
    }

    @Override
    public int read(ByteBuffer dst) throws IOException {
        //todo your code here
        init();
        for (Map.Entry<Integer, byte[]> entry : dataBlocksCache.entrySet()) {
                dst.put(entry.getValue());
        }
        return dst.remaining();
    }

    @Override
    public int write(ByteBuffer src) throws IOException {
        //todo your code here
        if (isReadOnly)
            throw  new ReadOnlyBufferException();
        int length = src.remaining();
        LocatedBlock block;
        if(fileSize==0)
        {
            while (src.hasRemaining()){
                block = nameNode.addBlock(uuid);
                blockAmount++;
                int size = src.remaining();
                byte[] bytes;
                if (size > DataNode.BLOCK_SIZE)
                    bytes= new byte[DataNode.BLOCK_SIZE];//需要写的内容
                else bytes = new byte[size];
                src.get(bytes);
                dataBlocksCache.put(block.getBlockNumber(),bytes);
            }
            fileSize+=length; // add fileSize
            return length;
        }
        //append
        int lastBid = getLastBid();
        int offset = dataBlocksCache.get(getLastBid()).length;
        int size = DataNode.BLOCK_SIZE - offset;//剩余空间
        byte[] bytes;
        if (size > src.remaining())
            bytes = new byte[src.remaining()];
        else  bytes = new byte[size];
        src.get(bytes);
        byte [] source = new byte[offset+bytes.length];//把原来的byte 与新的byte并起来
        System.arraycopy(dataBlocksCache.get(lastBid),0,source,0,offset);
        System.arraycopy(bytes,0,source,offset,bytes.length);
        dataBlocksCache.put(lastBid,source);

        while (src.hasRemaining()){
            block = nameNode.addBlock(uuid);
            blockAmount++;
            if (src.remaining() > DataNode.BLOCK_SIZE)
                bytes= new byte[DataNode.BLOCK_SIZE];//需要写的内容
            else bytes = new byte[src.remaining()];
            src.get(bytes);
            dataBlocksCache.put(block.getBlockNumber(),bytes);
        }

        fileSize+=length; // add fileSize
        return length;
    }

    @Override
    public long position() throws IOException {
        //todo your code here
        return 0;
    }

    @Override
    public SeekableByteChannel position(long newPosition) throws IOException {
        //todo your code here
        return null;
    }

    @Override
    public long size() throws IOException {
        //todo your code here
        int size =0;
        for (Map.Entry<Integer, byte[]> entry : dataBlocksCache.entrySet()) {
            size+=entry.getValue().length;
        }
        return size;
    }

    @Override
    public SeekableByteChannel truncate(long size) throws IOException {
        //todo your code here
        return null;
    }

    @Override
    public boolean isOpen() {
        //todo your code here
        return isOpen;
    }

    @Override
    public void close() throws IOException {
        //todo your code here
        //fileSize
        this.isOpen=false;
        fileNode.setFileSize(fileSize);
        nameNode.close();
    }

    @Override
    public void flush() throws IOException {
        //todo your code here
        try {
            DataNodeStub dataNodeStub = new DataNodeStub();
            for (Map.Entry<Integer, byte[]> entry : dataBlocksCache.entrySet()) {
                dataNodeStub.write(uuid,entry.getKey(),0,entry.getValue());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private int getLastBid(){
        Set<Integer> keySet = dataBlocksCache.keySet();
        Iterator<Integer>  iterator =keySet.iterator();
        int max =-1;
        while (iterator.hasNext()){
            int bid = iterator.next();
            if (max < bid ) max = bid;
        }
        return max;
    }
    private void init(){
        FileInputStream in=null;
        try{
            BlockInfo blocks = fileNode.getMainBlockInfo();
            Iterator<LocatedBlock> iterator = blocks.iterator();
            while(iterator.hasNext()){
                int bid = iterator.next().getBlockNumber();
                String path = "src/data/" + bid + ".block";
                File file = new File(path);
                in = new FileInputStream(file);
                byte [] bytes = new byte[in.available()];
                in.read(bytes);
                dataBlocksCache.put(bid,bytes);
                blockAmount++;
            }
            if (in!=null)
                in.close();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
}
